[テンプレ付き]PythonでCLIツールを作るときのTips

Pythonで、標準入力を変換して標準出力に出すCLIプログラムのテンプレートをご紹介します。

#Python

#初心者向け

#シェル

平野重利

2020.01.13

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、どんな作業もターミナルで行うことが多めの平野です。

最近はパイプに流すようなCLIアプリもPythonで作ることが多いので、そこで必要になったいくつかの要素をまとめてみます。

パイプライン処理として実装しよう
BrokenPipeの表示を消す
argparseによる引数とオプションのパース

この辺を考慮すれば、あとは文字列変換の主要なロジックだけを実装すればOKかと思います。

パイプライン処理として実装しよう

パイプライン処理とだけ言うと色々な意味がありそうですが、ここで言っている意味は データの先頭行の処理の結果は最終行が入力される前でも取り出せるようにしよう ということです。

パイプ (コンピュータ)#シェルからの使用 - Wikipedia

複数行のテキストが入力されてきた時に、それぞれの行の文字数をカウントするアプリケーションを作ったとします。この時、以下のようなスクリプトを書いてはいけません。

# ダメな例
lines = sys.stdin.readlines()
for line in lines:
    print(len(line))

このスクリプトの場合、最初のreadlines()をした時点で入力が全て入ってくるまで待つことになってしまいます。つまり出力側で最初の1行目が出力されるのは、入力側の最後の1行が入力された後になります。これは入力行数が多くない時には問題はありませんが、入力が非常に多くなってくるといつまでも結果が返ってこないので不便です。

また、それ以上に問題なのは、入力された文字列を全てメモリ上に保持しておく必要があることです。処理の対象ではないデータはメモリから消すことはスケーラブルなプログラムでは必須ですね。

パイプライン処理にする

以下のように、readline()（linesではなくline）で、 1行読み取っては処理、を繰り返すことでパイプライン処理となります。

line = sys.stdin.readline()
while line:
    print(len(line))
    line = sys.stdin.readline()

原理的に無理なものもある

もちろん何でもパイプラインにできるかというと、そんなことはないです。例えば各行をソートするプログラムは、最後の1行まで、その行がどこに入るのかわからないので、最初の1行すら出力することはできません。これはプログラム云々の話ではなく、不変の真理ってやつですね。

BrokenPipeの表示を消す

パイプラインは後続のコマンドによって、前段のコマンドを途中で終了させることがあります。大量の出力があるコマンドをheadに繋いだ場合が良い例です。

対策前

Pythonで単純に入力を出力に流すだけのcatもどきスクリプトを書いて、これを後続のheadで中断させてみます。

#!/usr/bin/env python
import os
import sys

line = sys.stdin.readline()
while line:
    line = line.strip("\n")
    print(line)
    line = sys.stdin.readline()

$ find / 2>/dev/null | python ~/cat.py | head
/
/home
/Developer
/Developer/MonoTouch
/Developer/MonoTouch/usr
/Developer/MonoTouch/usr/bin
/Developer/MonoTouch/usr/lib
/Developer/MonoTouch/usr/lib/mono
/Developer/MonoTouch/usr/lib/mono/2.1
/Developer/MonoTouch/usr/lib/mono/Xamarin.iOS
Traceback (most recent call last):
  File "/Users/hirano.shigetoshi/cat.py", line 8, in <module>
    print(line)
BrokenPipeError: [Errno 32] Broken pipe
Exception ignored in: <_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>
BrokenPipeError: [Errno 32] Broken pipe

BrokenPipeErrorというエラーが表示されてしまいました。この文言は標準エラー出力に出ているのでパイプで流す分には実害はありませんが、表示としてはとても邪魔です。

`BrokenPipeError`をハンドリングする

BrokenPipeErrorの消し方を調べてみると、

Note on SIGPIPE - signal --- 非同期イベントにハンドラを設定する

にハンドリング方法が記載されているので、そのまま以下のようなプログラムに修正します。

#!/usr/bin/env python
import os
import sys

try:
    line = sys.stdin.readline()
    while line:
        line = line.strip("\n")
        print(line)
        line = sys.stdin.readline()
except BrokenPipeError:
    devnull = os.open(os.devnull, os.O_WRONLY)
    os.dup2(devnull, sys.stdout.fileno())
    sys.exit(1)

$ find / 2>/dev/null | python ~/cat.py | head
/
/home
/Developer
/Developer/MonoTouch
/Developer/MonoTouch/usr
/Developer/MonoTouch/usr/bin
/Developer/MonoTouch/usr/lib
/Developer/MonoTouch/usr/lib/mono
/Developer/MonoTouch/usr/lib/mono/2.1
/Developer/MonoTouch/usr/lib/mono/Xamarin.iOS

これでBrokenPipeErrorが消えました。

`except`内でやっていること

except内の処理を見てみます。実は全然難しいことをやっていないです。

devnull = os.open(os.devnull, os.O_WRONLY)

os.devnullは普通は/dev/nullを表し、 os.O_WRONLYはwrite onlyのことなので、 open('/dev/null', 'w')という風にファイルを開いたのと同じと考えられます。

os.dup2(devnull, sys.stdout.fileno())

dup2はファイルディスクリプタを複製するOSのコマンドで、ここではdevnull(=/dev/null)をsys.stdout.fileno()(標準出力の番号)で複製するので、結局のところ、シェルで

1>/dev/null

と書くことに等しくなります（下にもうちょっと調べた記録を書きました）。

最後に終了コード0以外でプログラムを終了します。

sys.exit(1)

標準出力を捨てるように設定して終了する、という処理が書かれているだけのようです。実際この処理を入れず、単純にsys.exit(1)だけを実行した場合、後続のheadから中断がかかったとは言え、数行分の出力が標準出力のディスクリプタには送られてしまうようで、以下の文言が表示されました。

Exception ignored in: <_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>
BrokenPipeError: [Errno 32] Broken pipe

件の処理が入ると、標準出力に送られるはずのテキストを/dev/nullに送ったので、宛先不明のデータは存在せず上記エラーが出なくなるようです。

ファイルディスクリプタの複製について

完全に本題からズレますが、最近少し勉強したので備忘録として。

os.dup2(devnull, sys.stdout.fileno())

この指定で、devnullとstdoutの指定順が逆のように感じますが、もちろんこれで問題なく、以下のような動きになっているようです。

「 /dev/nullへ捨てるための出口（ファイルディスクリプタ）を複製して、標準出力へ出すことになっていた出口の番号（普通は1）をつける。標準出力へ出す予定だったテキストは出口番号1に送られるので、結局それは/dev/nullへ送られる。」

参考: dup, dup2

argparseによる引数とオプションのパース方法

Unixのいろいろなコマンドと同様に、-aとかでオプション指定させたいですね。 argparseはPython標準のライブラリで、コマンドラインからの引数や省略可能なオプションの指定をパースできます。

Argparse チュートリアル

以前はoptparseというライブラリを使っていましたが、このargparseはその後継にあたるもののようなので、今は何も考えずにargparseを使えば良さそうです。

使い方のおおよそ一覧

使い方は簡単なので、いくつかの使い方をまとめたプログラムをペタッとします。

import argparse
p = argparse.ArgumentParser()
p.add_argument('arg1', help='引数です')
p.add_argument('-a', '--opt_a', help='オプションです')
p.add_argument('-b', '--opt_b', default='aiueo', help='オプションです')
p.add_argument('--opt_c', default=0.1, type=float)
p.add_argument('-d', '--opt_d', action='store_true')
p.add_argument('-e', '--opt_e', action='store_true')
args = p.parse_args()

print(args.arg1)
if args.opt_a is not None:
    print(args.opt_a)
if args.opt_b is not None:
    print(args.opt_b)
if args.opt_c is not None:
    print(args.opt_c)
if args.opt_d is not None:
    print(args.opt_d)

p.add_argumentで受け取れる引数、オプションを追加していく
args.(引数名)かargs.(オプション名)でアクセスする
- オプション未指定時はNoneになる
-か--で始まるものがオプションで、それ以外は引数（指定必須）になる
- -(アルファベット一文字)の短縮形は省略可能
- default='hoge'でオプション未指定時のデフォルトを設定可能
- store_trueはオプション指定がある時Trueとする
  - store_trueのオプションを複数指定する場合、-deのようにまとめて指定可能
type=floatでfloat(args.opt_c)相当のキャストを行う
-hはヘルプを表示するオプションとして予約されている
- help='説明'の部分が表示される

非常に簡単で使いやすいです！複数オプションの短縮指定や、引数の後にオプションを指定できるなど、欲しい機能は全て網羅されている感じです。

テンプレート

ここまでの要素を合わせた、テンプレートは以下のような感じです。

#!/usr/bin/env python
import os
import sys
import argparse

p = argparse.ArgumentParser()
p.add_argument('arg1', help='引数です')
p.add_argument('-a', '--opt_a', help='オプションです')
p.add_argument('-b', '--opt_b', default='aiueo', help='オプションです')
p.add_argument('--opt_c', default=0.1, type=float)
p.add_argument('-d', '--opt_d', action='store_true')
p.add_argument('-e', '--opt_e', action='store_true')
args = p.parse_args()


def some_transform(line):
    return line


try:
    line = sys.stdin.readline()
    while line:
        line = line.strip("\n")
        line = some_transform(line)
        print(line)
        line = sys.stdin.readline()
except BrokenPipeError:
    devnull = os.open(os.devnull, os.O_WRONLY)
    os.dup2(devnull, sys.stdout.fileno())
    sys.exit(1)